package com.sui.bigdata.rtcadmin.service.impl;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.sui.bigdata.rtcadmin.constant.JobConstant;
import com.sui.bigdata.rtcadmin.dto.JobConfigDto;
import com.sui.bigdata.rtcadmin.exception.*;
import com.sui.bigdata.rtcadmin.repository.mapper.JobConfigMapper;
import com.sui.bigdata.rtcadmin.repository.model.JobConfig;
import com.sui.bigdata.rtcadmin.service.JobConfigService;
import lombok.NonNull;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.stream.Collectors;

import static com.sui.bigdata.rtcadmin.constant.JobConstant.FLINK_WEB_PREFIX;
import static com.sui.bigdata.rtcadmin.constant.JobConstant.FLINK_WEB_SUFFIX;

/**
 * @author YongChen
 * @date 2019/11/6 14:41
 * @description
 * @email yong_chen@sui.com
 */
@Service
public class JobConfigServiceImpl implements JobConfigService {

    @Autowired
    private JobConfigMapper jobConfigMapper;

    @Override
    public List<JobConfigDto> queryAll() {
        List<JobConfig> jobConfigs = jobConfigMapper.queryAll();
        jobConfigs.forEach((jobConfig -> jobConfig.setWebUrl(CustomConfiguration.getString(FLINK_WEB_PREFIX)+jobConfig.getAppId()+FLINK_WEB_SUFFIX)));
        return jobConfigs.stream().map(JobConfigDto :: buildJobConfigDto).collect(Collectors.toList());
    }

    @Override
    public JobConfigDto queryByAppName(@NonNull String jobName) {
        JobConfig jobConfig = jobConfigMapper.queryByJobName(jobName);
        if (jobConfig == null){
            return new JobConfigDto().setStatus(JobConstant.EDITED);
        }
        jobConfig.setWebUrl(CustomConfiguration.getString(FLINK_WEB_PREFIX) +jobConfig.getAppId()+ FLINK_WEB_SUFFIX);
        return JobConfigDto.buildJobConfigDto(jobConfig);
    }

    @Override
    public void save(@NonNull JobConfigDto jobConfigDto) throws Exception {
        if (!StringUtils.isNotBlank(jobConfigDto.getExecSql())) { throw new ExecSqlBlankException(); }
        if (!StringUtils.isNotBlank(jobConfigDto.getJobName())) { throw new JobNameBlankException(); }
        if (!StringUtils.isNotBlank(jobConfigDto.getDeveloper())) { throw new DeveloperBlankException();}

        JobConfig jobConfig = jobConfigMapper.queryByJobName(jobConfigDto.getJobName());
        if ( null != jobConfig && ("RUNNING".equals(jobConfig.getStatus()) || "ACCEPT".equals(jobConfig.getStatus()))) {
            throw new FlinkCommonException(jobConfigDto.getJobName() + " is "+jobConfig.getStatus()+" ,can't save!");
        }

        jobConfigMapper.saveAll(JobConfig.buildJobConfig(jobConfigDto));
        jobConfigMapper.updateStatusByJobName(jobConfigDto.getJobName(), JobConstant.EDITED, JobConstant.JOB_ONT_ONLINE);
    }

    @Override
    public void updateStatusByAppName(@NonNull String jobName,@NonNull String status, @NonNull int isOnline) {
        jobConfigMapper.updateStatusByJobName(jobName, status, isOnline);
    }

    @Override
    public void deleteByJobName(@NonNull String jobName) {
        JobConfig jobConfig = jobConfigMapper.queryByJobName(jobName);
        if(jobConfig == null){
            throw new FlinkJobConfigException(jobName + " not exist");
        }
        int isOnline = jobConfig.getIsOnline();
        if(isOnline == 1){
            throw new FlinkJobConfigException(jobName + " is online,please stop it first.");
        }
        jobConfigMapper.deleteByJobName(jobName);
    }

    @Override
    public void updateAlarmInfo(@NonNull JobConfigDto jobConfigDto){

        if (!StringUtils.isNotBlank(jobConfigDto.getJobName())) {
            throw new JobNameBlankException();
        }

        JobConfig jobConfig = jobConfigMapper.queryByJobName(jobConfigDto.getJobName());
        if (null == jobConfig) {
            throw new JobNotExistException(jobConfigDto.getJobName());
        }

        jobConfigMapper.updateAlarmByJobName(jobConfigDto.getJobName(),
                StringUtils.isNotBlank(jobConfigDto.getDeveloper()) ? jobConfigDto.getDeveloper() : jobConfig.getDeveloper(),
                StringUtils.isNotBlank(jobConfigDto.getAlarmGroupPhone()) ? jobConfigDto.getAlarmGroupPhone() : jobConfig.getAlarmGroupPhone(),
                StringUtils.isNotBlank(jobConfigDto.getDescription()) ? jobConfigDto.getDescription() : jobConfig.getDescription());

    }
}
